#include "catalog.h"
#include "query.h"
#include "sort.h"
#include "index.h"
#include <stdlib.h>

void copy_record(const Record source, Record &dest)
{
    dest.length = source.length;
    memcpy(dest.data, source.data, source.length);
}

void compare(Record left, int lOffset,  Record right, 
        int rOffset, Datatype type,  int &ans)
{
    void* lp = (void*)((char*)left.data + lOffset);
    void* rp = (void*)((char*)right.data + rOffset);
    // negative r, 0 or greater l
    if (type == STRING)
    {
        ans = (int) strcmp((char*)rp, (char*)lp);
    }
    else if(type == INTEGER)
    {
        int l = *(int*)lp;
        int r = *(int*)rp;
        if (l > r) ans = 1;
        else if (l == r) ans = 0;
        else ans = -1;
    }
    else
    {
        double l = *(double*)lp;
        double r = *(double*)rp;
        if (l > r) ans = 1;
        else if (l == r) ans = 0;
        else ans = -1;

    }
}
//
// SMJ: Sort merge join
// 
Status Operators::SMJ(const string & result,          // output relation name
        const int projCnt,              // number of attributes in the projection
        const AttrDesc attrDescArray[], // The projection list (as AttrDesc)
        const AttrDesc & attrDesc1,     // The left attribute in the join predicate
        const Operator op,              // The join operation
        const AttrDesc & attrDesc2,     // The right attribute in the join predicate
        const int reclen)               // The lenght of a tuple in the result relation
{
    cout << "Algorithm: SM Join" << endl;
    Status ret, lstat = OK, rstat = OK;
    HeapFile ansFile(result, ret);
    if (ret != OK) return ret;

    int numUnpinnedPages = bufMgr->numUnpinnedPages();
    int poolPages = (int)((.8) * numUnpinnedPages);
    int numBytes = poolPages * PAGESIZE;
    //how many tuples? poolpages * 1K / attrdesc1.attrlen
    int numTuples = numBytes /  attrDesc1.attrLen;
    if (numBytes % attrDesc1.attrLen)
        numTuples++;
    SortedFile sort1(attrDesc1.relName, attrDesc1.attrOffset, attrDesc1.attrLen,
            (Datatype)attrDesc1.attrType, numTuples, ret);
    if (ret != OK) return ret;
    SortedFile sort2(attrDesc2.relName, attrDesc2.attrOffset, attrDesc2.attrLen,
            (Datatype)attrDesc2.attrType, numTuples, ret);
    if (ret != OK) return ret;
    //get record from left, record from right
    //set marks
    //if equal add
    //if left is larger, advance R, vice versa
    Record recL, oldRecL, recR, oldRecR,  resultRec;
    int cmp = 0;
    lstat = sort1.next(recL);
    if (lstat != OK) return lstat;
    rstat = sort2.next(recR);
    sort2.setMark();
    if (rstat != OK) return rstat;
    oldRecL.data = malloc(recL.length);
    oldRecR.data = malloc(recR.length);
    copy_record(recL, oldRecL);
    copy_record(recR, oldRecR);
    while (rstat == OK && lstat == OK)
    {
        compare(recL, attrDesc1.attrOffset, recR, attrDesc2.attrOffset,
                (Datatype) attrDesc1.attrType, cmp);
        if (!cmp)
        {
            resultRec.length = reclen;
            resultRec.data = malloc(reclen);
            int recOffset = 0;
            for (int i = 0; i < projCnt; i++)
            {
                if (!strcmp(attrDescArray[i].relName, attrDesc1.relName))
                {
                    memcpy((char*)resultRec.data + recOffset,
                            (char*)recL.data + attrDescArray[i].attrOffset,
                            attrDescArray[i].attrLen);
                }
                else
                {
                    memcpy((char*)resultRec.data + recOffset,
                            (char*)recR.data + attrDescArray[i].attrOffset,
                            attrDescArray[i].attrLen);
                }
                recOffset += attrDescArray[i].attrLen;
            }


            RID resultRID;
            ret = ansFile.insertRecord(resultRec, resultRID);
            if (ret != OK) return ret;
            free(resultRec.data);
            
            //normally would just advance each
            //but...if R hasnt changed (match) dont go on
            //if it has, check if L has, if not, need to rollback
            rstat = sort2.next(recR);
            if (rstat != OK) break;
            compare(oldRecR, attrDesc2.attrOffset, 
                    recR, attrDesc2.attrOffset, 
                    (Datatype) attrDesc1.attrType, cmp);
            if (cmp != 0)
            {
                copy_record(recL, oldRecL);
                lstat = sort1.next(recL);
                if (lstat != OK) break;
                compare(recL, attrDesc1.attrOffset, 
                        oldRecL, attrDesc1.attrOffset, 
                        (Datatype) attrDesc1.attrType, cmp);
                if (cmp == 0)
                {
                    //L is the same, so go back to insert join
                    sort2.gotoMark();
                    rstat = sort2.next(recR);
                    if (rstat != OK) break;
                }
                else 
                {
                    sort2.setMark();
                    copy_record(recR, oldRecR);
                }
            }
        }
        else if (cmp > 0)
        {
            //l is larger so move r
            copy_record(recL, oldRecL);
            rstat = sort2.next(recR);
            sort2.setMark();
            if (rstat != OK) break; //eof
        }
        else
        {
            //r larger, move l
            copy_record(recL, oldRecL);
            lstat = sort1.next(recL);
            if (lstat != OK) break; //eof
        }
    }
    return OK;
}

